<?php

namespace teach\service;

use Yii;

class Kafka
{

    public function sendMsg($data)
    {
        $conf = new \RdKafka\Conf();
        $dir = "./kafka/" .  date("Y-m-d");
        if (!is_dir($dir)) {
            mkdir($dir, 0777, true);
        }
        $conf->setDrMsgCb(function ($kafka, $message) use($dir) {
            file_put_contents($dir . "/dr_cb.log", var_export($message, true) . PHP_EOL, FILE_APPEND);
        });
        $conf->setErrorCb(function ($kafka, $err, $reason) use ($dir) {
            file_put_contents($dir . "/err_cb.log", sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason) . PHP_EOL, FILE_APPEND);
        });

        $rk = new \RdKafka\Producer($conf);
        $rk->setLogLevel(LOG_DEBUG);
        $rk->addBrokers("127.0.0.1");

        $cf = new \RdKafka\TopicConf();
        // -1必须等所有brokers同步完成的确认 1当前服务器确认 0不确认，这里如果是0回调里的offset无返回，如果是1和-1会返回offset
        // 我们可以利用该机制做消息生产的确认，不过还不是100%，因为有可能会中途kafka服务器挂掉
        $cf->set('request.required.acks', 0);
        $topic = $rk->newTopic("msg_topic", $cf);

        $topic->produce(RD_KAFKA_PARTITION_UA, 0, $data);


        $len = $rk->getOutQLen();
        while ($len > 0) {
            $len = $rk->getOutQLen();
            //var_dump($len);
            $rk->poll(50);
        }
    }
}